package com.struggle.rabbitmqconsumer.listener.direct;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.rabbitmq.client.Channel;
import com.struggle.common.QueueConstants;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;

/**
 * @className DirectReceiver
 * @Description 消息接收监听类
 * @Author liuwj3
 * @Date 2019/10/12 0012 下午 2:13
 **/
@Component
// 监听队列的名称 test_direct_queue
@RabbitListener(queues = QueueConstants.DIRECT_QUEUE)
public class DirectReceiver implements ChannelAwareMessageListener {
    /*@RabbitHandler
    public void process(Map<String, Object> testMessage) throws JsonProcessingException {
        System.out.println("消费者收到消息：" + new ObjectMapper().writeValueAsString(testMessage));
    }*/

    /**
     * 消费者消息接收确认
     * @param message
     * @param channel
     * @throws Exception
     */
    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        System.out.println(QueueConstants.DIRECT_QUEUE + "队列的消费者收到消息");
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        try {
            /*String s = new String(message.getBody(), "UTF-8");
            Map<String, String> msgMap = new ObjectMapper().readValue(s, new TypeReference<Map<String, String>>() {
            });*/
            String s = message.toString().split("'")[1].trim();
            Map<String, String> msgMap = mapStringToMap(s);
            String messageId = msgMap.get("messageId");
            String messageData = msgMap.get("messageData");
            String createDate = msgMap.get("createDate");
            System.out.println("messageId:" + messageId + "  messageData:" + messageData + "  createDate:" + createDate);
            /**
             * multiple设置为true，表示支持批量确认机制以减少网络流量
             *
             * 例如：有值为5,6,7,8 deliveryTag的投递
             *
             * 如果此时channel.basicAck(8, true);则表示前面未确认的5,6,7投递也一起确认处理完毕。
             *
             * 如果此时channel.basicAck(8, false);则仅表示deliveryTag=8的消息已经成功处理。
             */
            channel.basicAck(deliveryTag, false);
        } catch (Exception e) {
            // 为true时消息会重新放回队列
            channel.basicReject(deliveryTag, false);
            e.printStackTrace();
        }


    }

    private Map<String, String> mapStringToMap(String str) {
        String[] split = str.substring(1, str.length() - 1).split(",");
        Map<String, String> map = new HashMap<>(8);
        for (String s : split) {
            String key = s.split("=")[0].trim();
            String value = s.split("=")[1];
            map.put(key, value);
        }
        return map;
    }
}
